package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.DwsTrafficForSourcePvBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.FlinkSourceUtil;
import com.atguigu.edu.realtime.dws.function.DimMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficVcSourceArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcSourceArIsNewPageViewWindow().start(
                11104,
                4,
                "traffic_vc_source_ar_is_new_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> pageLogDS) {
        //打印输入的page数据
        pageLogDS.print();
        //读取独立访客数据
        KafkaSource<String> uvTopicSource = FlinkSourceUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_UNIQUE_VISITOR_DETAIL, "traffic_vc_source_ar_is_new_page_view_window");
        DataStreamSource<String> uvTopicDS = env.fromSource(uvTopicSource,WatermarkStrategy.noWatermarks(),"uvTopicSource");
        //读取跳出用户数据
        KafkaSource<String> jumpTopicSource = FlinkSourceUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_USER_JUMP_DETAIL, "traffic_vc_source_ar_is_new_page_view_window");
        DataStreamSource<String> jumpTopicDS = env.fromSource(jumpTopicSource, WatermarkStrategy.noWatermarks(), "jumpTopicSource");
        //将三条流中的数据格式进行转换
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> pageBeanDS = pageLogDS.map(new MapFunction<String, DwsTrafficForSourcePvBean>() {
            @Override
            public DwsTrafficForSourcePvBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                Long ts = jsonObject.getLong("ts");

                return DwsTrafficForSourcePvBean.builder()
                        .versionCode(common.getString("vc"))
                        .sourceId(common.getString("sc"))
                        .ar(common.getString("ar"))
                        .isNew(common.getString("is_new"))
                        .uvCount(0L)
                        .totalSessionCount(page.getString("last_page_id") == null ? 1L : 0L)
                        .pageViewCount(1L)
                        .totalDuringTime(page.getLong("during_time"))
                        .jumpSessionCount(0L)
                        .ts(ts)
                        .build();
            }
        });
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> uvBeanDS = uvTopicDS.map(new MapFunction<String, DwsTrafficForSourcePvBean>() {
            @Override
            public DwsTrafficForSourcePvBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");

                return DwsTrafficForSourcePvBean.builder()
                        .versionCode(common.getString("vc"))
                        .sourceId(common.getString("sc"))
                        .ar(common.getString("ar"))
                        .isNew(common.getString("is_new"))
                        .uvCount(1L)
                        .totalSessionCount(0L)
                        .pageViewCount(0L)
                        .totalDuringTime(0L)
                        .jumpSessionCount(0L)
                        .ts(ts)
                        .build();
            }
        });
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> jumpBeanDS = jumpTopicDS.map(
                new MapFunction<String, DwsTrafficForSourcePvBean>() {
                    @Override
                    public DwsTrafficForSourcePvBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        Long ts = jsonObject.getLong("ts");
                        return DwsTrafficForSourcePvBean.builder()
                                .versionCode(common.getString("vc"))
                                .sourceId(common.getString("sc"))
                                .ar(common.getString("ar"))
                                .isNew(common.getString("is_new"))
                                .uvCount(0L)
                                .totalSessionCount(0L)
                                .pageViewCount(0L)
                                .totalDuringTime(0L)
                                .jumpSessionCount(1L)
                                .ts(ts)
                                .build();
                    }
                }
        );
        //合并三条流
        DataStream<DwsTrafficForSourcePvBean> unionDS = pageBeanDS.union(uvBeanDS).union(jumpBeanDS);
        //提取事件事件字段和水位线
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> withTimestampsAndWatermarks = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsTrafficForSourcePvBean>forBoundedOutOfOrderness(Duration.ofSeconds(15L))
                        .withTimestampAssigner(
                                (element, recordTimestamp) -> element.getTs()
                        )
        );
        //分组开窗
        WindowedStream<DwsTrafficForSourcePvBean, String, TimeWindow> windowDS = withTimestampsAndWatermarks.keyBy(
                new KeySelector<DwsTrafficForSourcePvBean, String>() {
                    @Override
                    public String getKey(DwsTrafficForSourcePvBean value) throws Exception {
                        return value.getVersionCode()
                                + value.getSourceId()
                                + value.getAr()
                                + value.getIsNew();
                    }
                }
        ).window(TumblingEventTimeWindows.of(Time.seconds(10L)));
        //聚合
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> reduceDS = windowDS.reduce(
                new ReduceFunction<DwsTrafficForSourcePvBean>() {
                    @Override
                    public DwsTrafficForSourcePvBean reduce(DwsTrafficForSourcePvBean value1, DwsTrafficForSourcePvBean value2) throws Exception {
                        value1.setTotalSessionCount(value1.getTotalSessionCount() + value2.getTotalSessionCount());
                        value1.setUvCount(value1.getUvCount() + value2.getUvCount());
                        value1.setTotalDuringTime(value1.getTotalDuringTime() + value2.getTotalDuringTime());
                        value1.setJumpSessionCount(value1.getJumpSessionCount() + value2.getJumpSessionCount());
                        value1.setPageViewCount(value1.getPageViewCount() + value2.getPageViewCount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<DwsTrafficForSourcePvBean, DwsTrafficForSourcePvBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<DwsTrafficForSourcePvBean, DwsTrafficForSourcePvBean, String, TimeWindow>.Context context, Iterable<DwsTrafficForSourcePvBean> elements, Collector<DwsTrafficForSourcePvBean> out) throws Exception {
                        TimeWindow window = context.window();
                        String start = DateFormatUtil.tsToDateTime(window.getStart());
                        String end = DateFormatUtil.tsToDateTime(window.getEnd());
                        for (DwsTrafficForSourcePvBean element : elements) {
                            element.setStt(start);
                            element.setEdt(end);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );
        //TODO 维度关联
        SingleOutputStreamOperator<DwsTrafficForSourcePvBean> dimBeanDS = reduceDS.map(new DimMapFunction());

        //将结果写入到doris中
        dimBeanDS.map(new BeanToJsonStrMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_source_ar_is_new_page_view_window values"));


    }
}
